#ifndef _TASK_RECV_H_
#define _TASK_RECV_H_

#include "../define/define.h"
#include "../trans_server/server.h"
#include "../trans_server/server_recv.h"
#include "../tcp_client/tc_man.h"
#include "./task_map.h"
#include "buf_map.h"

class task_recv : public tc_sink
{
public:
    
	task_recv()
	{
		m_log = new logfile("task_recv");
		m_client = new tc_man(this);
	}

    ~task_recv()
	{
		delete m_log;
		m_udp = 0;
	}

	void init(UDPServer* udp)
	{
	   m_udp = udp;
	}

	bool add_buf(ip_node task,uint32* d_socket,Link_info trans,uint32 con_id,
		char* buf, uint32 buf_size,uint32 s_socket,uint32 data_id)
	{
		m_log->WriteTrace("add buf");
		m_lock.EnterLock();

		uint32 time = get_time();
		ip_node_key node_key;
		node_key.d_ip = task.d_ip;
		node_key.d_port = task.d_port;
		node_key.l_port  = task.l_port;
		node_key.s_socket = s_socket;

        std::map<ip_node_key,uint32>::iterator it;
   
		it= m_map.find(node_key);
		if(it != m_map.end())
		{
			if(it->second != *d_socket)
			{
				//char ll[1024] ={0};
                //sprintf(ll,"new task,task && recv map error,s_socket=%d,socket=%d,d_socket=%d",
				//	s_socket,it->second,*d_socket);
				//m_log->WriteTrace(ll);
				m_lock.LeaveLock();
				return true;
			}

			std::map<uint32,server*>::iterator iter;
			iter = m_sendmap.find(*d_socket);
    
			if(iter != m_sendmap.end())
				iter->second->add_trans(task,trans,con_id);

			std::map<uint32,server_recv*>::iterator iter1;
			iter1 = m_recvmap.find(*d_socket);

			if(iter1 != m_recvmap.end())
			{
				iter1->second->put_data(buf,buf_size,data_id);
				char tmp_buf [1024] = {0};
				uint32 tmp_size = 1024;
				uint32 d_id = 0;
				iter1->second->get_data(tmp_buf,&tmp_size,&d_id);

				m_log->WriteTrace(tmp_buf);
				int size = m_client->send_buf(*d_socket,tmp_buf,tmp_size);
				
				char lle[1023] = {0};
				sprintf(lle,"send time=%d,time=%d,socket=%d,size=%d,tmp_size=%d",
					get_time() -time,get_time(),s_socket,size,tmp_size);
				m_log->WriteTrace(lle);

				if(size != tmp_size)
				{
					buf_st buf;
					buf.send = iter1->second;
					buf.socket = *d_socket;
					m_buf.add_map(buf);
				}

				if(size == 0)
				{
					m_lock.LeaveLock();
				    return true;
				}
									
				iter1->second->set_size(d_id,size);

				m_lock.LeaveLock();
				return true;
			}

			m_lock.LeaveLock();
			m_log->WriteTrace("new task,task && recv map error,map error");
			return false;
		}
		
		uint32 socket = 0;
		if(m_client->add_socket(&socket,task) == false)
		{
			m_lock.LeaveLock();
		    return false;
		}

		char lle[1023] = {0};
		sprintf(lle,"new socket time=%d,time=%d,socket=%d",
			get_time() -time,get_time(),s_socket);
		m_log->WriteTrace(lle);


		*d_socket = socket;
		server* send = m_sendlist.get_send(m_udp,s_socket,*d_socket,DATA_ack);

		send->add_trans(task,trans,con_id);
		
		char lls[1023] = {0};
		sprintf(lls,"new send time=%d,time=%d,socket=%d",
			get_time() -time,get_time(),s_socket);
		m_log->WriteTrace(lls);
		
		m_sendmap.insert(std::map<uint32,server*>::value_type(socket,send));
		
		server_recv* s_recv = m_recvlist.get_recv(*d_socket);
		m_recvmap.insert(std::map<uint32,server_recv*>::value_type(socket,s_recv));

		char llsr[1023] = {0};
		sprintf(llsr,"new recv time=%d,time=%d,socket=%d",
			get_time() -time,get_time(),s_socket);
		m_log->WriteTrace(llsr);
		ip_node_key insert_node_key;
		insert_node_key.d_ip = task.d_ip;
		insert_node_key.d_port = task.d_port;
		insert_node_key.l_port  = task.l_port;
		insert_node_key.s_socket = s_socket;

		m_map.insert(std::map<ip_node_key,uint32>::value_type(insert_node_key,socket));
		
		char ll[1024] = {0};
		in_addr ip; 
		memcpy(&ip,&task.d_ip,sizeof(in_addr));
		sprintf(ll,"new socket m_map.size=%d,s_socket=%d,d_socket=%d,d_ip=%s",
			m_map.size(),s_socket,socket,inet_ntoa(ip));
		m_log->WriteTrace(ll);

		s_recv->put_data(buf,buf_size,data_id);
		char tmp_buf [1024] = {0};
		uint32 tmp_size = 1024;
		uint32 d_id = 0;
		s_recv->get_data(tmp_buf,&tmp_size,&d_id);

		m_log->WriteTrace(tmp_buf);
		int size = m_client->send_buf(socket,tmp_buf,tmp_size);
		
		char llt[1023] = {0};
		sprintf(llt,"send time=%d,time=%d,socket=%d,size=%d,tmp_size=%d",
			get_time() -time,get_time(),s_socket,size,tmp_size);
		m_log->WriteTrace(llt);

		if(size != tmp_size)
		{
			buf_st buf;
			buf.send = s_recv;
			buf.socket = socket;
			m_buf.add_map(buf);
		}
		
		if(size == 0)
		{
			m_lock.LeaveLock();
			return true;
		}

		s_recv->set_size(d_id,size);
		m_lock.LeaveLock();
		return true;
	}

	bool add_trans(ip_node task,uint32 d_socket,uint32 my_socket)
	{
		m_lock.EnterLock();
		ip_node_key node_key;
		node_key.d_ip = task.d_ip;
		node_key.d_port = task.d_port;
		node_key.l_port  = task.l_port;
		node_key.s_socket = my_socket;

        std::map<ip_node_key,uint32>::iterator it;
        
		it= m_map.find(node_key);

		/*char ll[1024] = {0};
		in_addr ip; 
		memcpy(&ip,&task.d_ip,sizeof(uint32));
		sprintf(ll,"add trans:m_map.size=%d,socket=%d,node_key.d_port=%d,task.l_port=%d,d_ip=%s",
			m_map.size(),socket,task.d_port,task.l_port,inet_ntoa(ip));
		m_log->WriteTrace(ll);*/		

		if(it == m_map.end())
		{
			m_log->WriteTrace("add trans,can't find task && recv map error");
			m_lock.LeaveLock();
			return false;
		}

		if(it->second != my_socket)
		{
			m_log->WriteTrace("add trans,task && recv map error,socket error");
			m_lock.LeaveLock();
			return true;
		}

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(my_socket);
    
		if(iter == m_sendmap.end())
			return false;
	 
		for(uint32 i=0; i< task.count; i++)
		   iter->second->add_trans(task,task.trans_link[i],i);	
	    m_lock.LeaveLock();
	}
	
	bool ack_data(ip_node task, uint32 con_id,uint32 s_socket, uint32 d_socket,char* buf, uint32 buf_size)
	{
		m_lock.EnterLock();

		ip_node_key node_key;
		node_key.d_ip = task.d_ip;
		node_key.d_port = task.d_port;
		node_key.l_port  = task.l_port;
		node_key.s_socket = s_socket;

        std::map<ip_node_key,uint32>::iterator it;

		/*char ll[1024] = {0};
	    in_addr ip; 
		memcpy(&ip,&task.d_ip,sizeof(uint32));
		sprintf(ll,"m_map.size=%d,socket=%d,node_key.d_port=%d,task.l_port=%d,d_ip=%s",
			m_map.size(),s_socket,task.d_port,task.l_port,inet_ntoa(ip));
		m_log->WriteTrace(ll);*/
        
		it= m_map.find(node_key);

		if(it == m_map.end())
		{
			char ll[1024] = {0};
			sprintf(ll,"ack_data,can't find task && recv map error,socket=%d",s_socket);
			m_log->WriteTrace(ll);
			m_lock.LeaveLock();
			return false;
		}

		if(it->second != d_socket)
		{
			m_log->WriteTrace("ack_data,task && recv map error,socket error");
			m_lock.LeaveLock();
			return true;
		}

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(d_socket);
    
		if(iter == m_sendmap.end())
		{
			m_log->WriteTrace("ack data error");
			m_lock.LeaveLock();
			return false;
		}
	 
		m_lock.LeaveLock();
		return iter->second->ack_data(con_id,d_socket,buf,buf_size);
	}

	bool OnRead(uint32 socket, char* buf, uint32 buf_size)
	{	
		/*char ll[1024] = {0};
		sprintf(ll,"OnRead size=%d",buf_size);
		m_log->WriteTrace(ll);*/

		m_lock.EnterLock();
		if(buf_size == 0)
		{
			m_lock.LeaveLock();
			return true;
		}

		std::map<ip_node_key,uint32>::iterator it;
		for(it=m_map.begin();it!=m_map.end();it++)
		{
			if(it->second == socket)
			    break;
		}

		if(it->second != socket)
		{
			m_log->WriteTrace("read error not find task");
			m_lock.LeaveLock();
			return true;
		}
		
		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}

        int i;
		for(i=0; i< buf_size / 1024 ; i++)
			iter->second->Put_data(buf + i* 1024,1024);

		if(buf_size % 1024 > 0)
		    iter->second->Put_data(buf + i* 1024, buf_size - i*1024);
		m_lock.LeaveLock();
		return true;
	}


	bool Clear(uint32 socket)
	{
		m_lock.EnterLock();
		
		std::map<ip_node_key,uint32>::iterator it;
		for(it=m_map.begin();it!=m_map.end();it++)
		{
			if(it->second == socket)
			{
				char ll[1024] = {0};
				sprintf(ll,"delete m_map iter,m_map.size=%d,socket=%d,iter->socket=%d",
					m_map.size(),socket,it->second);
				m_log->WriteTrace(ll);
               	m_map.erase(it);
			    break;
			}
		}

		if(it == m_map.end())
		{
			m_log->WriteTrace("no this socket");
			m_lock.LeaveLock();
			return true;
		}

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_log->WriteTrace("delete server error");
			m_lock.LeaveLock();
			return false;
		}
        
		m_sendlist.free_send(iter->second);
		m_sendmap.erase(iter);
	
		std::map<uint32,server_recv*>::iterator iter1;
		iter1 = m_recvmap.find(socket);

		if(iter1 == m_recvmap.end())
		{
			m_log->WriteTrace("delete server_recv error");
			m_lock.LeaveLock();
			return false;
		}

		buf_st buf;
		buf.send = iter1->second;
		buf.socket = socket;

		m_buf.del_map(buf);
        
		m_recvlist.free_recv(iter1->second);
		m_recvmap.erase(iter1);	
		
		m_client->del_socket(socket);
       
        char lls[1024] ={0};
		sprintf(lls,"clear:socket=%d,is ok",socket);
		m_log->WriteTrace(lls);
		m_lock.LeaveLock();
		return true;
	}

	bool OnError(uint32 socket)
	{
		m_lock.EnterLock();
		
		char ll[1024] ={0};
		sprintf(ll,"OnError :socket=%d",socket);
		m_log->WriteTrace(ll);
		
		std::map<ip_node_key,uint32>::iterator it;
		for(it=m_map.begin();it!=m_map.end();it++)
		{
			if(it->second == socket)
			    break;
		}

		if(it == m_map.end())
			Clear(socket);

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_log->WriteTrace("delete server error");
			m_lock.LeaveLock();
			return false;
		}
       
		iter->second->set_big_id();

        char lls[1024] ={0};
		sprintf(lls,"down over :socket=%d",socket);
		m_log->WriteTrace(lls);
		m_lock.LeaveLock();
		return true;
	}

	bool OnBuffer(uint32 socket)
	{
		m_lock.EnterLock();

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}

		
		bool rt = iter->second->is_full();
        m_lock.LeaveLock();
		return rt;
	}

	bool OnSend()
	{
		m_lock.EnterLock();

		buf_st  send[1024];
		char    buf[1024];
        uint32  buf_size = 1024; 
		uint32  data_id;

		uint32 count = m_buf.get_map(send,1024);
  
		for(uint32 i=0; i<count; i++)
		{
			buf_size = 1024;

			int rt_size = send[i].send->get_data(buf,&buf_size,&data_id);

			if(rt_size == 0)
				continue;

			if(rt_size == 3)
			{
				m_buf.del_map(send[i]);
			    continue;
			}

			int size = m_client->send_buf(send[i].socket,buf,buf_size);
			
			if(size == 0)
                  continue;

			char lls[1024] = {0};
			sprintf(lls,"buf_size=%d,data_id=%d,socket=%d,send_size=%d,count=%d",
				buf_size,data_id,send[i].socket,size,count);
			m_log->WriteTrace(lls);

            send[i].send->set_size(data_id,size);
		}

		m_lock.LeaveLock();
        return true;
	}

private:

   tc_man*      m_client;
   UDPServer*   m_udp;

   logfile*     m_log;
   lock         m_lock;

   std::map<ip_node_key,uint32>                m_map;
   
   std::map<uint32,server*>                    m_sendmap;
   std::map<uint32,server_recv*>               m_recvmap;

   send_map                                    m_sendlist;
   recv_map                                    m_recvlist;

   buf_map                                     m_buf;
};

#endif
